package com.example.kafka;

import springfox.documentation.service.ApiListing;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.locks.ReentrantLock;

/**
 * Created by dong peng on 2019/11/16.
 */
public class KafkaTopicEvent {

    ExecutorService executorService = Executors.newFixedThreadPool(50);

    private static KafkaTopicEvent kafkaTopicEvent = new KafkaTopicEvent();
    private ReentrantLock lock = new ReentrantLock();

    private ConcurrentHashMap<String, Integer> topicVersion = new ConcurrentHashMap<>();

    public static KafkaTopicEvent getInstance() {
        return kafkaTopicEvent;
    }

    public void AddSubscribeEvent(TopicInfo topicInfo){
        // todo 直接计算并订阅
        topicVersion.put(topicInfo.getTopic(), 1);
        executorService.submit(new KafkaConsumerThread(1,topicVersion, topicInfo));
    }


    public void UpdateSubscribeEvent(TopicInfo topicInfo){
// todo 先停止原线程，再重新计算重新订阅 给版本号判断

        lock.lock();
        Integer v = topicVersion.get(topicInfo.getTopic());
        v += 1;
        topicVersion.put(topicInfo.getTopic(), v);
        lock.unlock();

        executorService.submit(new KafkaConsumerThread(v,topicVersion, topicInfo));
    }

    public void DelSubscribeEvent(TopicInfo topicInfo) {
        // TODO 停止线程

        lock.lock();
        Integer v = topicVersion.get(topicInfo.getTopic());
        v = v + 1;
        topicVersion.put(topicInfo.getTopic(), v);
        lock.unlock();
    }


    class KafkaConsumerThread implements Runnable {

        private Integer v;
        private ConcurrentHashMap<String, Integer>  currentTopicVersion;
        private TopicInfo topicInfo;

        KafkaConsumerThread(Integer version, ConcurrentHashMap CurrentTopicVersion, TopicInfo topicInfo) {
            this.v = version;
            this.currentTopicVersion = CurrentTopicVersion;
            this.topicInfo = topicInfo;
        }

        @Override
        public void run() {
            String topic = topicInfo.getTopic();
            Integer nowV = currentTopicVersion.get(topic);
            while (nowV.equals(v)) {

                System.out.println("轮询拉数, 线程名" + Thread.currentThread().getName() + "当前版本" + nowV);
                try {
                    Thread.sleep(3000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                nowV = currentTopicVersion.get(topic);
            }

            System.out.println(Thread.currentThread().getName() + "线程结束,主题名" + topic + "当前版本" + nowV + "之前版本" + v);

            executorService.shutdown();
        }
    }
}
